package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderRefund extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderRefund().start(10017,4, Constant.TOPIC_DWD_TRADE_ORDER_REFUND);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1.1 读取 topic_db
        readOdsDb(tEnv,groupId);

        // 1.2 读取 字典表
        createBaseDic(tEnv);

        // 2. 过滤退单表数据 order_refund_info   insert
        refundStream(tEnv);

        // 3. 过滤订单表中的退单数据: order_info  update //订单中 退单数据
        orderinfo_refund(tEnv);


        // 4. join: 普通的和 lookup join
//        tEnv.sqlQuery("select * from topic_db").execute().print();
        Table table = refund_join(tEnv);

        // 5. 写出到 kafka
        //创建 kafka结构
        create_refund(tEnv);

        //插入数据
        table.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_REFUND).execute();

    }

    private static void create_refund(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_REFUND + "\n" +
                "(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "province_id string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "refund_type string,\n" +
                "dic_name11 string,\n" +
                "refund_reason_type string,\n" +
                "dic_name22 string,\n" +
                "refund_reason_txt string,\n" +
                "refund_num string,\n" +
                "refund_amount string,\n" +
                "ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
    }

    private static Table refund_join(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("select\n" +
                "  ri.id,\n" +
                "  ri.user_id,\n" +
                "  ri.order_id,\n" +
                "  ri.sku_id,\n" +
                "  oi.province_id,\n" +
                "  date_format(ri.create_time,'yyyy-MM-dd') date_id,\n" +
                "  ri.create_time,\n" +
                "  ri.refund_type,\n" +
                "  dic1.info.dic_name as dic_name11,\n" +
                "  ri.refund_reason_type,\n" +
                "  dic2.info.dic_name as dic_name22,\n" +
                "  ri.refund_reason_txt,\n" +
                "  ri.refund_num,\n" +
                "  ri.refund_amount,\n" +
                "  ri.ts\n" +
                "from refund_info ri\n" +
                "join order_info oi\n" +
                "on ri.order_id =oi.id\n" +
                "join base_dic for system_time as of ri.pt as dic1\n" +
                "on ri.refund_type=dic1.rowkey\n" +
                "join base_dic for system_time as of ri.pt as dic2\n" +
                "on ri.refund_reason_type=dic2.rowkey");
    }

    private static void orderinfo_refund(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "  data['id'] id,\n" +
                "  data['province_id'] province_id,\n" +
                "  `old`\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='order_info'\n" +
                "and `type`='update'\n" +
                "and `old`['order_status'] is not null\n" +
                "and `data`['order_status'] ='1005'");

        tEnv.createTemporaryView("order_info",table);
    }

    private static void refundStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['refund_type'] refund_type,\n" +
                "data['refund_num'] refund_num,\n" +
                "data['refund_amount'] refund_amount,\n" +
                "data['refund_reason_type'] refund_reason_type,\n" +
                "data['refund_reason_txt'] refund_reason_txt,\n" +
                "data['create_time'] create_time,\n" +
                "proctime() pt,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='order_refund_info'\n" +
                "and `type`='insert'");

        tEnv.createTemporaryView("refund_info",table);
    }
}
